AI项目实战（9）一站式代码开发 Agent

"""项目六：数据模型层。

定义代码开发 Agent 的领域模型、配置常量和数据结构。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Optional

from core.config import DATA_DIR

# 工作空间根目录
WORKSPACE = DATA_DIR / "code_workspace"
WORKSPACE.mkdir(parents=True, exist_ok=True)

# 开发闭环配置
MAX_RETRY_ROUNDS = 3  # 最多自动修复 3 轮
TEST_TIMEOUT_SECONDS = 60  # 测试运行超时
MAX_FILE_CONTENT_LENGTH = 6000  # 文件读取最大长度


class TaskStatus(str, Enum):
    """任务状态。"""
    PENDING = "pending"
    ANALYZING = "analyzing"
    CODING = "coding"
    TESTING = "testing"
    FIXING = "fixing"
    SUCCESS = "success"
    FAILED = "failed"


@dataclass
class CodeFile:
    """代码文件模型。"""
    file_path: str
    content: str
    file_type: str = "python"  # python, test, config
    created_at: str = field(default="")


@dataclass
class TestResult:
    """测试结果模型。"""
    success: bool
    output: str
    failed_files: list[str] = field(default_factory=list)
    error_count: int = 0


@dataclass
class DevTask:
    """开发任务模型。"""
    task_id: str
    description: str
    status: TaskStatus = TaskStatus.PENDING
    files: list[CodeFile] = field(default_factory=list)
    test_results: list[TestResult] = field(default_factory=list)
    retry_count: int = 0
    final_report: str = ""

    def can_retry(self) -> bool:
        """判断是否还可以重试。"""
        return self.retry_count < MAX_RETRY_ROUNDS

    def add_file(self, file_path: str, content: str, file_type: str = "python") -> None:
        """添加代码文件。"""
        self.files.append(CodeFile(
            file_path=file_path,
            content=content,
            file_type=file_type,
        ))

    def add_test_result(self, success: bool, output: str) -> None:
        """添加测试结果。"""
        self.test_results.append(TestResult(
            success=success,
            output=output,
        ))

"""项目六：Prompt 层。

定义代码开发 Agent 的系统提示词和工具描述模板。
"""
from __future__ import annotations

# 主 Agent 系统提示词
SYSTEM_PROMPT = """你是全栈开发 Agent，负责从需求到代码的完整闭环。

## 核心能力
1. 需求分析：将自然语言需求转成结构化规格说明
2. 代码生成：生成符合规范的 Python 代码和测试
3. 测试运行：自动执行测试并收集结果
4. 错误修复：根据测试失败日志定位问题并修复

## 工作流程（严格按顺序）
1. **第一步**：调用 analyze_requirement 分析需求，明确接口和验收标准
2. **第二步**：基于分析结果，用 write_file 创建代码文件（先写测试，再写实现）
3. **第三步**：调用 run_tests 运行测试验证
4. **第四步**：如果测试失败：
   - 调用 read_file 查看相关代码
   - 定位问题根因
   - 调用 write_file 修复代码
   - 调用 run_tests 重新验证
5. 修复循环最多 3 轮，3 轮后仍失败则报告问题和你的分析
6. 全部通过后，汇报最终工作成果

## 编码规范
- Python 3.10+，使用类型注解
- 每个函数有 docstring，说明参数、返回值、异常
- 单文件不超过 300 行
- 测试和实现分开：test_xxx.py + xxx.py
- 使用 pytest 风格的测试用例
- 关键逻辑要有注释

## 文件命名约定
- 实现代码：{module_name}.py
- 测试代码：test_{module_name}.py
- 配置文件：config.py 或 settings.py

## 输出要求
- 每一步操作后简要说明做了什么
- 测试通过后列出所有创建的文件
- 如有修复，说明修改了什么和为什么
- 最终给出使用示例"""


# 需求分析工具的系统提示词
ANALYSIS_PROMPT = """你是需求分析师。请分析以下功能需求，输出结构化规格说明。

## 需要包含的内容
1. 功能描述：一句话概括这个功能做什么
2. 接口定义：函数签名、参数类型、返回值类型
3. 输入输出示例：至少 2 个示例
4. 边界条件：异常输入、空值、极限值的处理
5. 验收标准：什么样的结果算通过

## 输出格式
用 Markdown 格式，结构清晰，便于后续编码实现。"""


# 代码生成提示词模板
CODE_GEN_TEMPLATE = """根据以下需求生成高质量 Python 代码：

需求分析：
{analysis}

要求：
1. 类型注解完整
2. 有 docstring
3. 错误处理适当
4. 代码简洁可读

生成文件：{file_name}"""


# 测试生成提示词模板
TEST_GEN_TEMPLATE = """为以下代码生成 pytest 测试用例：

实现代码：
{code}

要求：
1. 正常情况测试
2. 边界条件测试
3. 异常情况测试
4. 测试用例有描述性名称

生成文件：test_{module_name}.py"""


# 错误修复提示词模板
FIX_PROMPT = """根据以下测试失败信息修复代码：

失败日志：
{error_log}

相关代码：
{code_content}

要求：
1. 定位问题根因
2. 给出修复方案
3. 保证不引入新问题
4. 保持原有接口不变"""

"""项目六：工具层。

定义代码开发 Agent 使用的所有工具函数。
"""
from __future__ import annotations

import os
import subprocess
from typing import Any

from langchain.tools import tool

from core.config import get_settings
from core.logging_conf import get_logger

from .models import (
    MAX_FILE_CONTENT_LENGTH,
    TEST_TIMEOUT_SECONDS,
    WORKSPACE,
)

logger = get_logger("p06.code_dev.tools")


@tool
def write_file(file_path: str, content: str) -> str:
    """将内容写入指定文件（在代码工作空间内）。

    Args:
        file_path: 相对于工作空间的路径，如 'app.py'、'utils/helpers.py'
        content: 文件的完整内容

    Returns:
        操作结果描述
    """
    try:
        full = WORKSPACE / file_path
        full.parent.mkdir(parents=True, exist_ok=True)
        full.write_text(content, encoding="utf-8")
        logger.info("写入文件: %s (%d 字节)", full, len(content))
        return f"✅ 文件已保存: {file_path} ({len(content)} 字节)"
    except Exception as e:
        logger.error("写入文件失败: %s", e)
        return f"❌ 写入失败: {e}"


@tool
def read_file(file_path: str) -> str:
    """读取工作空间中的文件内容。

    Args:
        file_path: 相对于工作空间的文件路径

    Returns:
        文件内容或错误信息
    """
    full = WORKSPACE / file_path
    try:
        content = full.read_text(encoding="utf-8")
        if len(content) > MAX_FILE_CONTENT_LENGTH:
            content = content[:MAX_FILE_CONTENT_LENGTH] + "\n... (内容过长，已截断)"
        return content
    except FileNotFoundError:
        return f"❌ 文件不存在: {file_path}"
    except Exception as e:
        logger.error("读取文件失败: %s", e)
        return f"❌ 读取失败: {e}"


@tool
def list_workspace() -> str:
    """列出代码工作空间中所有文件。

    Returns:
        工作空间文件列表
    """
    try:
        files = []
        for p in sorted(WORKSPACE.rglob("*")):
            if p.is_file() and ".git" not in str(p) and "__pycache__" not in str(p):
                rel = p.relative_to(WORKSPACE)
                size = p.stat().st_size
                files.append(f"  {rel} ({size} 字节)")
        if not files:
            return "📂 工作空间为空，尚无文件。"
        result = "📂 工作空间文件：\n" + "\n".join(files[:50])
        if len(files) > 50:
            result += f"\n  ... 还有 {len(files) - 50} 个文件"
        return result
    except Exception as e:
        logger.error("列出文件失败: %s", e)
        return f"❌ 列出文件失败: {e}"


@tool
def run_tests(test_path: str = ".") -> str:
    """运行指定路径的 Python 测试。

    Args:
        test_path: 测试目录或文件路径（相对于工作空间），默认为当前目录

    Returns:
        测试运行结果
    """
    full = WORKSPACE / test_path
    try:
        result = subprocess.run(
            ["python", "-m", "pytest", str(full), "-x", "--tb=short", "-q"],
            capture_output=True,
            text=True,
            timeout=TEST_TIMEOUT_SECONDS,
            cwd=str(WORKSPACE),
            env={**os.environ, "PYTHONPATH": str(WORKSPACE)},
        )
        output = result.stdout
        if len(output) > MAX_FILE_CONTENT_LENGTH:
            output = output[-MAX_FILE_CONTENT_LENGTH:] + "\n... (输出过长，已截断末尾)"

        if not output.strip():
            output = result.stderr[-2000:] if len(result.stderr) > 2000 else result.stderr

        if result.returncode == 0:
            return f"✅ 测试通过！\n\n{output or '无输出'}"
        else:
            return f"❌ 测试失败（退出码: {result.returncode}）\n\n{output or '无输出'}"
    except subprocess.TimeoutExpired:
        return f"⏱️ 测试运行超时（{TEST_TIMEOUT_SECONDS} 秒），请检查是否有死循环或慢速测试"
    except Exception as e:
        logger.error("测试运行失败: %s", e)
        return f"❌ 测试运行失败: {e}"


@tool
def analyze_requirement(description: str) -> str:
    """分析功能需求，输出结构化规格说明。

    包括：接口定义（函数签名）、输入输出类型、边界条件、验收标准。

    Args:
        description: 用户的功能需求描述

    Returns:
        结构化的需求分析
    """
    return (
        f"📋 需求分析\n"
        f"原始描述：{description}\n\n"
        f"（Agent 将基于此分析进行接口设计和编码实现）\n\n"
        f"建议分析方向：\n"
        f"1. 功能概述：这个函数/模块主要做什么？\n"
        f"2. 输入参数：需要哪些参数？类型是什么？\n"
        f"3. 返回值：返回什么数据？结构如何？\n"
        f"4. 边界条件：空值、异常值、极限值如何处理？\n"
        f"5. 验收标准：什么情况算功能正确？"
    )


def get_all_tools() -> list[Any]:
    """获取所有可用工具列表。

    Returns:
        工具对象列表
    """
    return [
        analyze_requirement,
        write_file,
        read_file,
        list_workspace,
        run_tests,
    ]

"""项目六：服务层。

封装代码开发 Agent 的核心业务逻辑，包括任务管理、流程编排等。
"""
from __future__ import annotations

from typing import Any

from core import build_chat_model
from core.logging_conf import get_logger

from .models import DevTask, TaskStatus
from .prompts import SYSTEM_PROMPT
from .tools import get_all_tools

logger = get_logger("p06.code_dev.service")


class CodeDevService:
    """代码开发服务类。

    封装代码开发 Agent 的核心业务逻辑，包括：
    - 创建开发任务
    - 跟踪任务状态
    - 管理开发闭环
    - 生成最终报告
    """

    def __init__(self) -> None:
        self._tasks: dict[str, DevTask] = {}
        self._agent: Any | None = None

    def create_task(self, description: str, task_id: str | None = None) -> DevTask:
        """创建新的开发任务。

        Args:
            description: 功能需求描述
            task_id: 可选的任务 ID，自动生成

        Returns:
            任务对象
        """
        import uuid

        if task_id is None:
            task_id = f"dev_{uuid.uuid4().hex[:8]}"

        task = DevTask(
            task_id=task_id,
            description=description,
            status=TaskStatus.PENDING,
        )
        self._tasks[task_id] = task
        logger.info("创建开发任务: %s", task_id)
        return task

    def get_task(self, task_id: str) -> DevTask | None:
        """获取任务。

        Args:
            task_id: 任务 ID

        Returns:
            任务对象或 None
        """
        return self._tasks.get(task_id)

    def update_task_status(self, task_id: str, status: TaskStatus) -> bool:
        """更新任务状态。

        Args:
            task_id: 任务 ID
            status: 新状态

        Returns:
            是否成功
        """
        task = self.get_task(task_id)
        if task is None:
            logger.warning("任务不存在: %s", task_id)
            return False
        task.status = status
        logger.info("任务 %s 状态更新为: %s", task_id, status)
        return True

    def increment_retry(self, task_id: str) -> bool:
        """增加任务重试计数。

        Args:
            task_id: 任务 ID

        Returns:
            是否成功（达到最大重试次数时返回 False）
        """
        task = self.get_task(task_id)
        if task is None:
            return False
        if not task.can_retry():
            logger.warning("任务 %s 已达到最大重试次数", task_id)
            return False
        task.retry_count += 1
        return True

    def generate_final_report(self, task_id: str) -> str:
        """生成任务最终报告。

        Args:
            task_id: 任务 ID

        Returns:
            报告内容
        """
        task = self.get_task(task_id)
        if task is None:
            return "任务不存在"

        lines = [
            "# 开发任务完成报告",
            "",
            f"**任务 ID**: {task.task_id}",
            f"**需求描述**: {task.description}",
            f"**最终状态**: {task.status.value}",
            "",
            "## 创建的文件",
            "",
        ]

        if task.files:
            for f in task.files:
                lines.append(f"- `{f.file_path}` ({f.file_type})")
        else:
            lines.append("（无文件创建记录）")

        lines.extend([
            "",
            "## 测试结果",
            "",
        ])

        if task.test_results:
            for i, result in enumerate(task.test_results, 1):
                status = "✅ 通过" if result.success else "❌ 失败"
                lines.append(f"第 {i} 轮: {status}")
        else:
            lines.append("（无测试运行记录）")

        if task.retry_count > 0:
            lines.extend([
                "",
                "## 修复记录",
                "",
                f"自动修复 {task.retry_count} 轮",
            ])

        lines.extend([
            "",
            "## 使用说明",
            "",
            "所有文件已保存到工作空间目录。",
            "如需进一步修改，请提供新的需求描述。",
        ])

        report = "\n".join(lines)
        task.final_report = report
        return report

    def build_agent(self) -> Any:
        """构建代码开发 Agent。

        Returns:
            LangChain Agent 对象
        """
        if self._agent is not None:
            return self._agent

        from langchain.agents import create_agent

        self._agent = create_agent(
            model=build_chat_model(),
            tools=get_all_tools(),
            system_prompt=SYSTEM_PROMPT,
        )
        return self._agent

    def run_dev_task(self, description: str) -> str:
        """运行完整的开发任务流程。

        Args:
            description: 功能需求描述

        Returns:
            任务执行结果
        """
        task = self.create_task(description)
        self.update_task_status(task.task_id, TaskStatus.ANALYZING)

        agent = self.build_agent()

        try:
            result = agent.run(description)
            self.update_task_status(task.task_id, TaskStatus.SUCCESS)
            return result
        except Exception as e:
            logger.error("开发任务执行失败: %s", e)
            self.update_task_status(task.task_id, TaskStatus.FAILED)
            return f"开发任务执行失败: {e}"


# 全局服务实例
_service: CodeDevService | None = None


def get_service() -> CodeDevService:
    """获取代码开发服务实例。

    Returns:
        CodeDevService 单例
    """
    global _service
    if _service is None:
        _service = CodeDevService()
    return _service

"""项目六：项目定义层。

定义项目注册和对外暴露的接口。
"""
from __future__ import annotations

from typing import Any

from core import BaseProject, registry

from .service import get_service
from .tools import get_all_tools


class CodeDevProject(BaseProject):
    """一站式代码开发 Agent 项目。

    主讲能力：复杂 Harness 闭环、测试驱动开发、自动修复循环

    业务场景：用户提出一个功能需求，Agent 自动完成需求分析、接口设计、
    代码生成、测试编写、测试运行、失败修复与最终交付。
    """

    id = "p06_code_dev"
    name = "一站式代码开发 Agent"
    description = "需求→设计→编码→测试→修复→验证，完整开发闭环。"
    capabilities = ["Harness", "Tool", "Test-Driven"]

    def build_agent(self) -> Any:
        """构建 Agent 实例。

        Returns:
            LangChain Agent 对象
        """
        service = get_service()
        return service.build_agent()

    def run(self, message: str) -> str:
        """运行项目任务。

        Args:
            message: 用户输入的需求描述

        Returns:
            Agent 响应
        """
        service = get_service()
        return service.run_dev_task(message)


# 项目实例
project = CodeDevProject()

# 对外暴露的快捷函数
run = project.run
build_agent = project.build_agent
get_tools = get_all_tools

__all__ = [
    "CodeDevProject",
    "project",
    "run",
    "build_agent",
    "get_tools",
]

"""项目六：一站式代码开发 Agent（复杂 Harness 闭环）

主讲能力：复杂 Harness 闭环——需求分析 → 代码生成 → 测试验证 → 修复 → 再测试

业务场景：用户描述功能需求，Agent 自动完成完整开发闭环：
分析需求 → 设计接口 → 生成代码 → 编写测试 → 运行测试 → 根据失败修复 → 重新验证

生产级特性：
- StateGraph 驱动的开发闭环（非简单一问一答）
- 需求分析和接口设计先于编码（避免返工）
- 测试驱动：先生成测试用例，再生成实现代码
- 自动修复循环：测试失败 → 分析原因 → 修复 → 重跑，最多 3 轮
- Git 友好：每次修改生成清晰的 commit message
- 文件操作安全：不覆盖用户未授权的文件

架构分层：
- models.py: 数据模型、配置常量
- prompts.py: 系统提示词、模板
- tools.py: 工具函数
- service.py: 业务逻辑、Agent 编排
- project.py: 项目定义、对外接口
"""
from __future__ import annotations

from core import registry

# 导出 models 层
from .models import (
    CodeFile,
    DevTask,
    MAX_FILE_CONTENT_LENGTH,
    MAX_RETRY_ROUNDS,
    TaskStatus,
    TestResult,
    TEST_TIMEOUT_SECONDS,
    WORKSPACE,
)

# 导出 prompts 层
from .prompts import (
    ANALYSIS_PROMPT,
    CODE_GEN_TEMPLATE,
    FIX_PROMPT,
    SYSTEM_PROMPT,
    TEST_GEN_TEMPLATE,
)

# 导出 tools 层（保持向后兼容）
from .tools import (
    analyze_requirement,
    get_all_tools,
    list_workspace,
    read_file,
    run_tests,
    write_file,
)

# 导出 service 层
from .service import CodeDevService, get_service

# 导出 project 层
from .project import (
    CodeDevProject,
    build_agent,
    get_tools,
    project,
    run,
)

# 注册项目到全局注册表
registry.register(project)

__all__ = [
    # models
    "WORKSPACE",
    "MAX_RETRY_ROUNDS",
    "TEST_TIMEOUT_SECONDS",
    "MAX_FILE_CONTENT_LENGTH",
    "TaskStatus",
    "CodeFile",
    "TestResult",
    "DevTask",
    # prompts
    "SYSTEM_PROMPT",
    "ANALYSIS_PROMPT",
    "CODE_GEN_TEMPLATE",
    "TEST_GEN_TEMPLATE",
    "FIX_PROMPT",
    # tools
    "write_file",
    "read_file",
    "list_workspace",
    "run_tests",
    "analyze_requirement",
    "get_all_tools",
    # service
    "CodeDevService",
    "get_service",
    # project
    "CodeDevProject",
    "project",
    "run",
    "build_agent",
    "get_tools",
]